#include "pch.h"
#include "ImportData.h"

std::vector<Field> ImportData::columns_ex_ = { Field("VAULT_ID", "VARCHAR", 64, false, "", "", ""),
                                                       Field("SIGN_VAULT_ID", "VARCHAR", 64, false, "", "", ""),
                                                       Field("VALUES_HASH", "CHAR", 64, false, "", "PRI", ""),
                                                       Field("TIME_STAMP", "TIMESTAMP", 0, true, "CURRENT_TIMESTAMP", "", "")};


ImportData::ImportData(std::shared_ptr<EnclaveProvider> enclave_provider, Config &config)
	: writer_(nullptr), enclave_provider_(enclave_provider), config_(config)
{
}


ImportData::~ImportData()
{
}

void ImportData::importData(const::enclave::ImportDataRequest * request, ::grpc::ServerWriter<::enclave::ImportDataResponse>* writer)
{
	LOG(INFO) << "Load import cache file";

	import_data_cache_.load();

    faild_cache_datas_ = import_data_cache_.faild_cache_datas();
	success_cache_pos_ = import_data_cache_.success_cache_pos();

	data_des_ = new HanaData(config_.database_);

	if (data_des_->hdbc_ == nullptr)
	{
		LOG(ERROR) << "Hana connect error";
		return;
	}

	writer_ = writer;

    import_data_config_.database_.host_ = request->host();
    import_data_config_.database_.port_ = request->port();
    import_data_config_.database_.database_ = request->database();
    import_data_config_.database_.user_ = request->user();
    import_data_config_.database_.password_ = request->password();
	import_data_config_.tables_.clear();
	
	data_src_ = new MysqlData(import_data_config_.database_);
	if (data_src_->mysql_ == nullptr)
	{
		LOG(ERROR) << "Mysql connect error";
		return;
	}

	if (request->tables_size() == 0)
	{
		std::vector<std::string> tables;
		data_src_->showTables(tables);

		for (auto iter = tables.begin(); iter != tables.end(); ++iter)
		{
			TableConfig table;
			table.name_ = *iter;

			import_data_config_.tables_.push_back(table);
		}
	}
	else {
		for (int i = 0; i < request->tables_size(); ++i)
		{
			TableConfig table;
			table.name_ = request->tables(i).name();
			table.order_by_field_ = request->tables(i).orderbyfield();

			for (int j = 0; j < request->tables(i).fields_size(); ++j)
			{
				Field field;
				field.name_ = request->tables(i).fields(j).name();
				table.fields_.push_back(field);
			}

			for (int j = 0; j < request->tables(i).opefields_size(); ++j)
			{
				Field field;
				field.name_ = request->tables(i).opefields(j).name();
				table.ope_fields_.push_back(field);
			}

			import_data_config_.tables_.push_back(table);
		}
	}

	for (auto i = 0; i < import_data_config_.tables_.size(); ++i)
	{
		TableConfig &table = import_data_config_.tables_[i];

		LOG(INFO) << "Table name: " << table.name_ << ", " << "Columns count" << table.fields_.size();
		
		LOG(INFO) << "Counting row ...";

		std::vector<Field> columns;

		int row_count = data_src_->num_rows(table.name_);
		tables_count_[table.name_] = row_count;

		LOG(INFO) << "Row count = " << row_count;

		columns.clear();
		data_src_->showColumns(table.name_, columns);
		tables_imported_count_[table.name_] = 0;

		if (table.fields_.size() > 0)
		{
			for (auto column_iter = columns.begin(); column_iter != columns.end();)
			{
				auto iter = std::find(table.fields_.begin(), table.fields_.end(), Field(column_iter->name_));

				if (iter == table.fields_.end()) {
					column_iter = columns.erase(column_iter);
				}
				else {
					++column_iter;
				}
			}
		}

		char buf[1024] = { 0 };
		for (auto column_iter = columns.begin(); column_iter != columns.end(); ++column_iter)
		{
			if (column_iter->type_ == "datetime"
				|| column_iter->type_ == "date" 
				|| column_iter->type_ == "int"
				|| column_iter->type_ == "tinyint"
				|| column_iter->type_ == "timestamp")
			{
				column_iter->type_ = "DECIMAL";
				column_iter->length_ = 38;
			}
			else if (std::find(table.ope_fields_.begin(), table.ope_fields_.end(), *column_iter) != table.ope_fields_.end())
			{
				column_iter->type_ = "DECIMAL";
				column_iter->length_ = 38;
			}
			else 
			{
				column_iter->type_ = "varchar";
				column_iter->length_ = enclave_provider_->encrypt_size(column_iter->length_);

				if (column_iter->length_ > 5000)
				{
					column_iter->length_ = 5000;
				}
			}

			if (column_iter->key_ == "PRI")
			{
				column_iter->key_ = "";

				if (table.order_by_field_.empty())
				{
					table.order_by_field_ = column_iter->key_;
				}
			}

//			memset(buf, 0, sizeof(buf));
//			enclave_provider_.encrypt(column_iter->name_.c_str(), (int)column_iter->name_.size(), buf, sizeof(buf));
//
//			column_iter->name_ = buf;
		}

		columns.insert(columns.end(), columns_ex_.begin(), columns_ex_.end());

		if (data_des_->isTableExists(table.name_) == false)
		{
			LOG(INFO) << "Creating table " << table.name_;

			data_des_->createTable(table.name_, columns);

			LOG(INFO) << "Created table " << table.name_;
		}

		tables_columns_[table.name_] = columns;
	}

	delete data_des_;
	data_des_ = nullptr;

	for (int i = 0; i < config_.thread_number_; i++)
	{
		auto import_data_thread = std::shared_ptr<ImportDataThread>(new ImportDataThread(this, config_.database_, enclave_provider_, config_));

		if (import_data_thread->run())
		{
			import_threads_.push_back(import_data_thread);
		}
	}

	LOG(INFO) << "Import number of threads is " << import_threads_.size();

	HANDLE *handle = new HANDLE[import_threads_.size()];
	for (size_t i = 0; i < import_threads_.size(); ++i)
	{
		handle[i] = import_threads_[i]->thread();
	}
	WaitForMultipleObjects((int)import_threads_.size(), handle, TRUE, INFINITE);

    delete[]handle;
	delete data_src_;
	data_src_ = nullptr;
	
	table_index_ = 0;
	tables_count_.clear();
	tables_imported_count_.clear();
    tables_columns_.clear();

	tables_pos_.clear();

	import_threads_.clear();

	enclave::ImportDataResponse response;
	grpc::WriteOptions write_options;
	writer_->WriteLast(response, write_options);
	writer_ = nullptr;
}

bool ImportData::getTableFields(TableConfig **table_config, std::vector<Field> **columns, std::vector<Field> &selected_columns, IData::Fields &fields, int &rows_num, ImportDataCache::Data &pos)
{
	CriticalSectionUtil cs_util = CriticalSectionUtil(critical_section_);

	if (table_index_ >= tables_count_.size())
	{
		return false;
	}

	*table_config = &(import_data_config_.tables_[table_index_]);
	*columns = &(tables_columns_[(*table_config)->name_]);

	int table_count = tables_count_[(*table_config)->name_];

	ImportDataCache::Data &table_pos = tables_pos_[(*table_config)->name_];

	if (faild_cache_datas_.find((*table_config)->name_) != faild_cache_datas_.end() && faild_cache_datas_[(*table_config)->name_].size() > 0)
	{
		table_pos = faild_cache_datas_[(*table_config)->name_].front();
		
		faild_cache_datas_[(*table_config)->name_].erase(faild_cache_datas_[(*table_config)->name_].begin());

		LOG(INFO) << "Import cache table = " << (*table_config)->name_  << "filed_index = " << table_pos.field_index_ << ", pos = " << table_pos.pos_ << ", length = " << table_pos.length_;

		if (faild_cache_datas_[(*table_config)->name_].size() == 0)
		{
			faild_cache_datas_.erase((*table_config)->name_);

			if (success_cache_pos_.find((*table_config)->name_) == success_cache_pos_.end())
			{
				LOG(FATAL) << "Import cached error";
				return false;
			}

			LOG(INFO) << "Import cache table = " << (*table_config)->name_ << " end, set success pos = " << table_pos.pos_;
		}
	}
	else if (success_cache_pos_.find((*table_config)->name_) != success_cache_pos_.end())
	{
		table_pos = success_cache_pos_[(*table_config)->name_];

		success_cache_pos_.erase((*table_config)->name_);
	}

	if (table_pos.length_ == 0)
	{
		table_pos.length_ = config_.read_count_;
	}

	char temp[64] = { 0 };
	_itoa_s(table_pos.length_, temp, 10);

	std::string fields_str;
	if ((*table_config)->fields_.size() > 0)
	{
		for (auto iter = (*table_config)->fields_.begin(); iter != (*table_config)->fields_.end(); iter ++)
		{
			if (iter != (*table_config)->fields_.begin())
			{
				fields_str += ", ";

				fields_str += iter->name_;
			}
		}
	} 
	else
	{
		fields_str += "*";
	}

	std::string sql = "SELECT " + fields_str +" FROM " + (*table_config)->name_ + " where " + (*table_config)->order_by_field_ + " >= '" + table_pos.field_index_ + "' ORDER BY " + (*table_config)->order_by_field_ + " LIMIT " + std::string(temp) + ";";

	if (data_src_->select(sql, fields, rows_num, &selected_columns) == false)
	{
		LOG(ERROR) << "select error: " << sql;
		return true;
	}

	pos = table_pos;

	if (success_cache_pos_.find((*table_config)->name_) == success_cache_pos_.end())
	{
		if (rows_num > 0)
		{
			int offset = 0;
			int column_index = 0;
			int column_len = 0;
			for (size_t i = selected_columns.size() - 1; i >= 0; --i)
			{
				column_len = fields.field_len(fields.field_index_ - 1 - column_index);
				offset += column_len;

				++column_index;

				if (selected_columns[i] == (*table_config)->order_by_field_)
				{
					break;
				}
			}

			table_pos.field_index_ = std::string(fields.buf() + fields.buf_pos_ - offset, column_len);

			addFaild((*table_config)->name_.c_str(), pos);
		}
		
		if (rows_num < table_pos.length_) {

			table_index_++;
		}
	}

    return true;
}

void ImportData::importProgress(ImportDataThread * thread, TableConfig& table_config, int length)
{
	CriticalSectionUtil cs_util = CriticalSectionUtil(imported_count_cs_);

	tables_imported_count_[table_config.name_] += length;
	int imported_count = tables_imported_count_[table_config.name_];
    
	enclave::ImportDataResponse response;
	auto table = response.add_tables();
	table->set_name(table_config.name_);
	table->set_datacount(tables_count_[table_config.name_]);
	table->set_importeddatacount(imported_count);
    writer_->Write(response);
}

void ImportData::addFaild(const char * table, ImportDataCache::Data & data)
{
	import_data_cache_.addFaild(table, data, tables_pos_[table]);
}

void ImportData::removeFaild(const char * table, ImportDataCache::Data & data)
{
	CriticalSectionUtil cs_util = CriticalSectionUtil(critical_section_);

	import_data_cache_.removeFaild(table, data);
}